package com.central.utils

import java.lang
import java.util.Date

import com.central.bean.FocusPersonDetail
import com.central.car.streaming.SaveEsCarData
import com.central.face.streaming.SaveEsData
import com.redislabs.provider.redis._
import com.redislabs.provider.redis.{RedisConfig, RedisEndpoint}
import redis.clients.jedis.Protocol
//import com.central.utils.RedisUtils.redisConfig
import org.apache.spark.Partitioner
import org.apache.spark.rdd.RDD
import org.json4s.NoTypeHints
import org.json4s.jackson.JsonMethods.parse
import org.json4s.jackson.Serialization
import redis.clients.jedis.{JedisPool, JedisPoolConfig}
import scala.reflect.ClassTag
import scala.util.{Failure, Success, Try}
import scala.reflect.runtime.universe.TypeTag

object RedisUtil {

  val host: String = ConfigUtil.properties.getProperty("redis.host")
  val password: String = ConfigUtil.properties.getProperty("redis.password")
  val port: Integer = ConfigUtil.properties.getProperty("redis.port").toInt
  val wifiCodeEcrelationDbname: Integer = ConfigUtil.properties.getProperty("wifi.code.ecrelation.data.name").toInt
  val redisConfig = new RedisConfig(RedisEndpoint(host, port, password, Protocol.DEFAULT_DATABASE, Protocol.DEFAULT_TIMEOUT))
  private val jedisPoolConfig = new JedisPoolConfig
  jedisPoolConfig.setMaxTotal(1000)

  private val jedisPool = new JedisPool(jedisPoolConfig, host, port, 3000, password, 2)
  private val jedisPoolHash = new JedisPool(jedisPoolConfig, host, port, 3000, password, 6)
  private val jedisCarPool = new JedisPool(jedisPoolConfig, host, port, 3000, password, 3)
  private val jedisResidentPool = new JedisPool(jedisPoolConfig, host, port, 3000, password, 4)

  def accomulatorByString(string: String, face_captime: String, codeType: String, savedata: SaveEsData, code_captime: String): (Boolean, String, String, String, String, SaveEsData, String) = {
    var flag: Boolean = false
    var count: String = "0"
    var key: String = string
    val jedis = jedisPool.getResource
    val long: lang.Long = jedis.incr(string)
    if (long != 1) {

      count = long + ""
      if (long == 5) {
        jedis.persist(string)
      }
      if (long > 5) {
        flag = true
      }
    }
    else {
      //      jedis.set(string, "1")
      jedis.expire(string, 30 * 24 * 60 * 60)
      count = "1"
    }
    jedis.close()
    return (flag, key, count, face_captime, codeType, savedata, code_captime)
  }


  def accomulatorByHash(string: String, face_captime: String, codeType: String, savedata: SaveEsData, code_captime: String): (Boolean, String, String, String, String, SaveEsData, String) = {
    var flag: Boolean = false
    var count: String = "0"
    val key = string.split("_")(0)
    val hkey = string.split("_")(1)
    val jedis = jedisPoolHash.getResource

    val long = jedis.hincrBy(key, hkey, 1)
    count = long + ""
    if (long > 5) {
      flag = true
    }

    jedis.close()

    return (flag, string, count, face_captime, codeType, savedata, code_captime)
  }


  /**
    * 车码关联计数
    *
    * @param string
    * @param face_captime
    * @param codeType
    * @param savedata
    * @param code_captime
    * @return
    */
  def accomulatorCarByString(string: String, face_captime: String, codeType: String, savedata: SaveEsCarData, code_captime: String): (Boolean, String, String, String, String, SaveEsCarData, String) = {
    var flag: Boolean = false
    var count: String = "0"
    var key: String = string
    val jedis = jedisCarPool.getResource
    val long: lang.Long = jedis.incr(string)
    if (long != 1) {

      count = long + ""
      if (long == 5) {
        jedis.persist(string)
      }
      if (long > 5) {
        flag = true
      }
    }
    else {
      //      jedis.set(string, "1")
      jedis.expire(string, 30 * 24 * 60 * 60)
      count = "1"
    }
    jedis.close()

    return (flag, key, count, face_captime, codeType, savedata, code_captime)
  }


  def accomulatorByHash(string: String): Boolean = {
    var flag: Boolean = false


    return flag

  }

  def loadDataFromRedis[T](rdd: RDD[T], key: String): RDD[(String, String)] = {

    val value: RDD[(String, String)] = rdd.sparkContext.fromRedisHash(key)(redisConfig)
      .filter(kv => {
        !(kv._1 == null || kv._1.equals(""))
      })
    value
  }


  def transRedisRdd(redisRdd: RDD[(String, String)]) = {
    redisRdd
      .mapPartitions(
        partitionRdds => partitionRdds.map {
          case (id, jDetail) => { //将json转换成对应的对象,其中jDetail为:[{xxxx},{xxx}]
            implicit val formats = Serialization.formats(NoTypeHints)
            val details = parse(jDetail).children.map(jvalue => jvalue.extractOpt[FocusPersonDetail]) //这里添加了对群体人员的支持
            (id, details)
          }
        })
      .combineByKey[List[FocusPersonDetail]]( //过滤预警时间超过范围的数据
      (detailOpts: List[Option[FocusPersonDetail]]) => {
        detailOpts.filter(_.exists(detail => {
          val now = new Date().getTime
          detail.begintime <= now && detail.endtime >= now
        })).map(_.get)
      },
      (details: List[FocusPersonDetail], detailOpts: List[Option[FocusPersonDetail]]) => {
        val dtails = detailOpts.filter(_.exists(detail => {
          val now = new Date().getTime
          detail.begintime <= now && detail.endtime >= now
        })).map(_.get)
        details ++: dtails
      },
      (details1: List[FocusPersonDetail], details2: List[FocusPersonDetail]) => details1 ++: details2,
      Partitioner.defaultPartitioner(redisRdd),
      true
    )

  }

  def cleanTheWarnedRecords[T <: Product : ClassTag : TypeTag](warningRecords: RDD[T])(fieldsInMd5: T => String): Either[RDD[T], Throwable] = {
    import com.redislabs.provider.redis._
    val recordsWithMd5 = warningRecords
      .map(m => (fieldsInMd5(m), m))
      .groupByKey()
      .map(recs => (recs._1, recs._2.head)) //如果记录内部有重复,先排除掉
    val warnedRecords = Try(loadDataFromRedis(recordsWithMd5, "RS_WARNED_RECORDS")).getOrElse(warningRecords.sparkContext.emptyRDD[(String, String)])
    val newWarningRecords = recordsWithMd5.leftOuterJoin(warnedRecords).filter(_._2._2.isEmpty).cache()
    //将排重后结果保存到redis当中
    Try {
      //      val redisConfig = new RedisConfig(new RedisEndpoint(host.get, port.get.toInt,null,Protocol.DEFAULT_DATABASE,Protocol.DEFAULT_TIMEOUT))  //new RedisEndpoint(host, port,null,Protocol.DEFAULT_DATABASE,Protocol.DEFAULT_TIMEOUT)
      warningRecords.sparkContext.toRedisHASH(newWarningRecords.map(r => (r._1, (new Date().getTime).toString)), "RS_WARNED_RECORDS")(redisConfig)
    } match {
      case Success(_) => {
        val toSaveDatas = newWarningRecords.map(_._2._1) //得到排重后可以保存的数据
        Left(toSaveDatas)
      }
      case Failure(err) => Right(new RuntimeException(s"保存Redis缓存库时出错::${err.getMessage}"))
    }
  }

  /**
    * 码码关联计数
    **/
  def checkAndUpdateWifiRelation(src_code: String, tag_code: String, duration: Int): Integer = {
    val checkOne = tag_code + "_" + src_code
    val checkTwo = src_code + "_" + tag_code
    var flag: Integer = 1;
    val jedis = jedisPool.getResource
    jedis.select(wifiCodeEcrelationDbname)
    val bool = jedis.exists(checkOne)
    if (bool) {
      flag = jedis.incr(checkOne).intValue()
      if (flag == duration) {
        jedis.persist(checkOne)
      }
      flag = flag * -1
    } else {
      flag = jedis.incr(checkTwo).intValue()
      if (flag == duration) {
        jedis.persist(checkTwo)
      } else if (flag == 1) {
        jedis.expire(checkTwo, 30 * 24 * 60 * 60)
      }
    }
    jedis.close()
    flag
  }

  /**
    *
    * kafka update offset
    **/

  def updateOffset(topic: String, count: Long) = {
    val jedis = jedisPool.getResource
    jedis.select(wifiCodeEcrelationDbname)
    jedis.incrBy(topic, count)
  }


  /**
    *
    * kafka get offset
    **/

  def getOffset(topic: String): Long = {
    val jedis = jedisPool.getResource
    jedis.select(wifiCodeEcrelationDbname)
    val count = jedis.get(topic)
    if (count == null) {
      return 0
    } else {
      return count.toLong
    }
  }

  /**
    * 常驻人口计数
    *
    * @param string "侦码"+"_"+"时间",type
    * @return
    */
  def accomulatorResident(string: (String, String)): (Boolean, String, String) = {
    var flag: Boolean = false
    val resource = jedisResidentPool.getResource
    val code = string._1.split("_")(0)
    val time = string._1.split("_")(1)
    val long = ParseTime.DateToLong(time) / 1000 / 60 / 60 / 24

    val booleanDay = resource.incr(code + "_"+long)

    if (booleanDay==1) {
      val code_count = resource.incr(code)
      if (code_count > 5) {
        flag = true
      }
      if (code_count == 1) {
        resource.expire(code, 30 * 24 * 60 * 60 )
      }
      resource.expire(code +"_"+ long, 24 * 60 * 60 )
    }

    resource.close()
    return (flag, code, string._2)

  }

  def main(args: Array[String]): Unit = {

    val resource = jedisPool.getResource
    val jedis3 = jedisCarPool.getResource
    val long = jedis3.hincrBy("dsas", "sdasad", 1)
    //    val long = jedis3.incr("sadadasd")
    println(long)
  }
}
